#include "EventLoop.h"

// 通过线程局部存储防止一个线程创建多个EventLoop
__thread EventLoop *t_loopInThisThread = nullptr;
// 定义Poller默认的IO复用接口的超时时间
const int kPollTimeMs = 10000;

// 创建wakeupfd，用来唤醒subReactor处理新来的channel
int createEventfd()
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_FATAL("eventfd error:%d %s\n", errno, strerror(errno));
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false)
    , quit_(false)
    , callingPendingFunctors_(false)
    , threadId_(CurrentThread::tid())
    , poller_(Poller::newDefaultPoller(this))
    , wakeupFd_(createEventfd())
    , wakeupChannel_(new Channel(this, wakeupFd_))
{
    LOG_DEBUG("EventLoop created %p in thread %d\n", this, threadId_);
    if(t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d\n", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    // 设置wakeupfd的读事件回调
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 每一个EventLoop都需要监听wakeupChannel的EPOLLIN读事件
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

// 开启事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;

    LOG_INFO("EventLoop %p start looping \n", this);

    while(!quit_)
    {
        activeChannels_.clear(); // 清除已处理完的活跃的Channel
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); // epoll_wait
        for(Channel* channel : activeChannels_)
        {
            // Poller监听到事件的发生，然后上报给EventLoop，EventLoop通知Channel处理对应的事件
            channel->handleEvent(pollReturnTime_);
        }
        // 待标注
        doPendingFunctors();
    }

    LOG_INFO("EventLoop %p stop looping. \n", this);
    looping_ = false;
}

// 退出事件循环，quit函数可能由当前事件循环所在的线程执行
// 也可能是其他线程执行的，如果是其他线程执行的就需要通过
// wakeupFd来唤醒要退出事件循环的线程
void EventLoop::quit()
{
    quit_ = true;
    if(!isInLoopThread())
    {
        wakeup();
    }
}

// 在当前的loop执行回调cb
void EventLoop::runInLoop(Functor cb)
{
    if(isInLoopThread()) // 在当前的loop线程中执行cb回调
    {
        cb();
    }
    else // 不是在当前的loop线程中执行cb回调
    {
        queueInLoop(std::move(cb)); 
    }
}

// 将cb放入队列中，唤醒loop所在的线程，执行cb回调
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    // 唤醒对应的loop线程来执行回调操作
    // || callingPendingFunctors_的意思是：当前loop正
    // 在执行回调，但是loop又有了新的回调
    if(!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup(); // 唤醒对应的loop线程
    }
}

// 用来唤醒loop所在的线程的
// 向wakeupFd_写一个数据，wakeupChannel就
// 会发生读事件，当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if(n != sizeof one)
    {
        LOG_ERROR("EventLoop::wakeup writes %lu bytes instead of 8\n", n);
    }
}

// 当read读取到数据，其所在的loop线程就会被唤醒
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if(n != sizeof one)
    {
        LOG_ERROR("EventLoop::handleRead reads %lu bytes instead of 8\n", n);
    }
}

void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for(const Functor& functor : functors)
    {
        functor(); // 当前loop线程执行需要执行的回调操作
    }

    callingPendingFunctors_ = false;
}

// 以下的方法都是调用Poller的方法
void EventLoop::updateChannel(Channel* channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
    return poller_->hasChannel(channel);
}